import datetime
import getopt
import json
import sys

from dateutil.relativedelta import relativedelta

from air_web.config.config import config
from air_web.data_platform import mysql_con

task_base_save_table = config.get("ORGNO_TYPEID_15MIN", "orgno_typeid_15min")
task_low_save_table = config.get("LOW_ORGNO_TYPEID_15MIN", "low_orgno_typeid_15min")
task_baseline_save_table = config.get(
    "BASELINE_ORGNO_TYPEID_15MIN", "baseline_orgno_typeid_15min"
)
orderly_power_15min = config.get("ORDERLY_POWER_15MIN", "orderly_power_15min")
demand_power_15min = config.get("DEMAND_POWER_15MIN", "demand_power_15min")

task_base = {
    "step_base": {
        "type": "base_agg",
        "module": "air_web.dw.dws_15min",
        "class": "AggBase",
        "is_baseline": False,
    },
    "step_on7_type": {
        "type": "agg",
        "dimensions": ["on7", "type_id", "data_time"],
        "rename": {"on7": "org_no"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_pare": {
        "type": "agg",
        "dimensions": ["on7", "pare_type_id", "data_time"],
        "rename": {"on7": "org_no", "pare_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_sort": {
        "type": "agg",
        "dimensions": ["on7", "type_code_sort", "data_time"],
        "rename": {"on7": "org_no", "type_code_sort": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_all": {
        "type": "agg",
        "dimensions": ["on7", "highest_type_id", "data_time"],
        "rename": {"on7": "org_no", "highest_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_type": {
        "type": "agg",
        "dimensions": ["on5", "type_id", "data_time"],
        "rename": {"on5": "org_no"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_pare": {
        "type": "agg",
        "dimensions": ["on5", "pare_type_id", "data_time"],
        "rename": {"on5": "org_no", "pare_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_sort": {
        "type": "agg",
        "dimensions": ["on5", "type_code_sort", "data_time"],
        "rename": {"on5": "org_no", "type_code_sort": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_all": {
        "type": "agg",
        "dimensions": ["on5", "highest_type_id", "data_time"],
        "rename": {"on5": "org_no", "highest_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_type": {
        "type": "agg",
        "dimensions": ["province_no", "type_id", "data_time"],
        "rename": {"province_no": "org_no"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_pare": {
        "type": "agg",
        "dimensions": ["province_no", "pare_type_id", "data_time"],
        "rename": {"province_no": "org_no", "pare_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_sort": {
        "type": "agg",
        "dimensions": ["province_no", "type_code_sort", "data_time"],
        "rename": {"province_no": "org_no", "type_code_sort": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_all": {
        "type": "agg",
        "dimensions": ["province_no", "highest_type_id", "data_time"],
        "rename": {"province_no": "org_no", "highest_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
}
task_low = {
    "step_base": {
        "type": "base_agg",
        "module": "air_web.dw.dws_15min",
        "class": "AggLow",
        "is_baseline": False,
    },
    "step_on7_type": {
        "type": "agg",
        "dimensions": ["on7", "type_id", "data_time"],
        "rename": {"on7": "org_no"},
        "save_table": task_low_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_pare": {
        "type": "agg",
        "dimensions": ["on7", "pare_type_id", "data_time"],
        "rename": {"on7": "org_no", "pare_type_id": "type_id"},
        "save_table": task_low_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_sort": {
        "type": "agg",
        "dimensions": ["on7", "type_code_sort", "data_time"],
        "rename": {"on7": "org_no", "type_code_sort": "type_id"},
        "save_table": task_low_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_type": {
        "type": "agg",
        "dimensions": ["on5", "type_id", "data_time"],
        "rename": {"on5": "org_no"},
        "save_table": task_low_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_pare": {
        "type": "agg",
        "dimensions": ["on5", "pare_type_id", "data_time"],
        "rename": {"on5": "org_no", "pare_type_id": "type_id"},
        "save_table": task_low_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_sort": {
        "type": "agg",
        "dimensions": ["on5", "type_code_sort", "data_time"],
        "rename": {"on5": "org_no", "type_code_sort": "type_id"},
        "save_table": task_low_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_type": {
        "type": "agg",
        "dimensions": ["province_no", "type_id", "data_time"],
        "rename": {"province_no": "org_no"},
        "save_table": task_low_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_pare": {
        "type": "agg",
        "dimensions": ["province_no", "pare_type_id", "data_time"],
        "rename": {"province_no": "org_no", "pare_type_id": "type_id"},
        "save_table": task_low_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_sort": {
        "type": "agg",
        "dimensions": ["province_no", "type_code_sort", "data_time"],
        "rename": {"province_no": "org_no", "type_code_sort": "type_id"},
        "save_table": task_low_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
}
task_industry_agg = {
    "step_base": {
        "type": "industry_base_agg",
        "module": "air_web.dw.dws_industry_15min",
        "class": "IndResAggBase",
        "source_table": task_base_save_table,
    },
    "step_industry_agg": {
        "type": "step_industry_agg",
        "module": "air_web.dw.dws_industry_15min",
        "class": "IndResAgg",
        "dimensions": ["org_no", "industry_id", "data_time"],
        "rename": {"industry_id": "type_id"},
        "save_table": task_base_save_table,
        "is_valid": 1,
    },
    "step_all_ind_agg": {
        "type": "step_industry_agg",
        "module": "air_web.dw.dws_industry_15min",
        "class": "IndResAgg",
        "dimensions": ["org_no", "all_industry_id", "data_time"],
        "rename": {"all_industry_id": "type_id"},
        "save_table": task_base_save_table,
        "is_valid": 1,
    },
}

task_max_time = {
    "step_max_time": {
        "type": "max_time",
        "module": "air_web.dw.dws_time",
        "class": "MaxTime",
        "rename": {
            "p_total_sum": "max_p_total",
            "p_kt_sum": "max_p_kt",
            "p_base_sum": "max_p_base",
        },
        "source_table": "orgno_typeid_15min",
        "save_table": "area_max_time",
        "is_valid": 1,
    }
}

task_baseline = {
    "step_base": {
        "type": "base_agg",
        "module": "air_web.dw.dws_15min",
        "class": "AggBase",
        "is_baseline": True,
    },
    "step_on7_type": {
        "type": "agg",
        "dimensions": ["on7", "type_id", "data_time"],
        "rename": {"on7": "org_no"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_pare": {
        "type": "agg",
        "dimensions": ["on7", "pare_type_id", "data_time"],
        "rename": {"on7": "org_no", "pare_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_sort": {
        "type": "agg",
        "dimensions": ["on7", "type_code_sort", "data_time"],
        "rename": {"on7": "org_no", "type_code_sort": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_all": {
        "type": "agg",
        "dimensions": ["on7", "highest_type_id", "data_time"],
        "rename": {"on7": "org_no", "highest_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_type": {
        "type": "agg",
        "dimensions": ["on5", "type_id", "data_time"],
        "rename": {"on5": "org_no"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_pare": {
        "type": "agg",
        "dimensions": ["on5", "pare_type_id", "data_time"],
        "rename": {"on5": "org_no", "pare_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_sort": {
        "type": "agg",
        "dimensions": ["on5", "type_code_sort", "data_time"],
        "rename": {"on5": "org_no", "type_code_sort": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_all": {
        "type": "agg",
        "dimensions": ["on5", "highest_type_id", "data_time"],
        "rename": {"on5": "org_no", "highest_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_type": {
        "type": "agg",
        "dimensions": ["province_no", "type_id", "data_time"],
        "rename": {"province_no": "org_no"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_pare": {
        "type": "agg",
        "dimensions": ["province_no", "pare_type_id", "data_time"],
        "rename": {"province_no": "org_no", "pare_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_sort": {
        "type": "agg",
        "dimensions": ["province_no", "type_code_sort", "data_time"],
        "rename": {"province_no": "org_no", "type_code_sort": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_all": {
        "type": "agg",
        "dimensions": ["province_no", "highest_type_id", "data_time"],
        "rename": {"province_no": "org_no", "highest_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
}


task_org_day = {
    "step_org_day_stat": {
        "type": "org_day_stat",
        "module": "air_web.dw.dws_day",
        "class": "OrgDayStat",
        "source_table": "orgno_typeid_15min",
        "save_table": "orgno_typeid_day_stat",
        "is_valid": 1,
    }
}
task_org_type_day = {
    "step_org_type_day_stat": {
        "type": "org_type_day_stat",
        "module": "air_web.dw.dws_day",
        "class": "OrgTypeDayStat",
        "source_table": "orgno_typeid_15min",
        "save_table": "orgno_typeid_ymd_stat",
        "is_valid": 1,
    }
}

# 四川配置
task_day_sichuan = {
    "step_city_report": {
        "type": "city_report",
        "module": "air_web.dw.city_report",
        "class": "AggCityReport",
        "source_table": "orgno_typeid_15min",
        "save_table": "city_report",
        "is_valid": 1,
    },
    "step_city_alert": {
        "type": "city_alert",
        "module": "air_web.dw.dws_day",
        "class": "GetCityAlert",
        "source_table": "",
        "save_table": "city_alert",
        "is_valid": 1,
    },
    "step_city_control": {
        "type": "city_control",
        "module": "air_web.dw.dws_day",
        "class": "CityControl",
        "source_table": "",
        "save_table": "city_control",
        "is_valid": 1,
    },
}

task_year = {
    "step_area_cons_num": {
        "type": "stat_cons_num",
        "module": "air_web.dw.stat_cons_num",
        "class": "StatSonsNum",
        "save_table": "area_cons_num",
        "is_valid": 1,
    }
}

# 四川配置
task_cons_day = {
    "step_cons_day": {
        "type": "file",
        "module": "air_web.dw.cons_day_stat",
        "class": "AggConsDay",
        "save_table": "cons_day_stat",
        "is_valid": 1,
    }
}

task_cons_day_stat = {
    "task_cons_day_stat": {
        "type": "file",
        "module": "air_web.dw.cons_day_stat",
        "class": "ConsDayStat",
        "save_table": "cons_ymd_stat",
        "is_valid": 1,
    }
}

task_first = {
    "step_init_mysql": {
        "type": "init",
        "module": "air_web.dw.init_mysql",
        "class": "InitTable",
        "is_valid": 0,
    },
    "step_gen_partition": {
        "type": "init",
        "module": "air_web.dw.gen_partition",
        "class": "GenTablePartion",
        "is_valid": 0,
    },
    "step_init": {
        "type": "init",
        "module": "air_web.dw.dws_first",
        "class": "InitJudge",
        "is_valid": 1,
    },
    "step_cons": {
        "type": "proc_cons",
        "module": "air_web.dw.dws_first",
        "class": "CCons",
        "save_table": "c_cons",
        "is_valid": 1,
    },
    "step_merge_cons": {
        "type": "proc_cons",
        "module": "air_web.dw.dws_first",
        "class": "MergeCons",
        "save_table": "merge_cons",
        "is_valid": 1,
    },
    "step_area_type_map": {
        "type": "proc_cons",
        "module": "air_web.dw.dws_first",
        "class": "AreaTypeMap",
        "save_table": "area_type_map",
        "is_valid": 1,
    },
}

task_modify_alias = {
    "step_modify_alias": {
        "type": "file",
        "module": "air_web.dw.modify_alias",
        "class": "ModifyAlias",
        "is_valid": 1,
    }
}


task_cons_sort = {
    "step_cons_sort": {
        "type": "file",
        "module": "air_web.dw.cons_sort",
        "class": "ConsSort",
        "is_valid": 1,
    }
}

task_orderly = {
    "step_base": {
        "type": "base_agg",
        "module": "air_web.dw.spec_cons_15min",
        "class": "SpecConsAggBase",
        "flag_field": "orderly_flag",
    },
    "step_on7_pare": {
        "type": "spec_cons_agg",
        "dimensions": ["on7", "pare_type_id", "measure_code", "data_time"],
        "rename": {"on7": "org_no", "pare_type_id": "type_id"},
        "save_table": orderly_power_15min,
        "is_valid": 1,
    },
    "step_on5_pare": {
        "type": "spec_cons_agg",
        "dimensions": ["on5", "pare_type_id", "measure_code", "data_time"],
        "rename": {"on5": "org_no", "pare_type_id": "type_id"},
        "save_table": orderly_power_15min,
        "is_valid": 1,
    },
    "step_pro_pare": {
        "type": "spec_cons_agg",
        "dimensions": [
            "province_no",
            "pare_type_id",
            "measure_code",
            "data_time",
        ],
        "rename": {"province_no": "org_no", "pare_type_id": "type_id"},
        "save_table": orderly_power_15min,
        "is_valid": 1,
    },
}
task_demand = {
    "step_base": {
        "type": "base_agg",
        "module": "air_web.dw.spec_cons_15min",
        "class": "SpecConsAggBase",
        "flag_field": "demand_flag",
    },
    "step_on7_pare": {
        "type": "spec_cons_agg",
        "dimensions": ["on7", "pare_type_id", "measure_code", "data_time"],
        "rename": {"on7": "org_no", "pare_type_id": "type_id"},
        "save_table": demand_power_15min,
        "is_valid": 1,
    },
    "step_on5_pare": {
        "type": "spec_cons_agg",
        "dimensions": ["on5", "pare_type_id", "measure_code", "data_time"],
        "rename": {"on5": "org_no", "pare_type_id": "type_id"},
        "save_table": demand_power_15min,
        "is_valid": 1,
    },
    "step_pro_pare": {
        "type": "spec_cons_agg",
        "dimensions": [
            "province_no",
            "pare_type_id",
            "measure_code",
            "data_time",
        ],
        "rename": {"province_no": "org_no", "pare_type_id": "type_id"},
        "save_table": demand_power_15min,
        "is_valid": 1,
    },
}


def main(
    task_name,
    timedelta,
    config_text,
    task_rank,
    start_time=None,
    end_time=None,
    depend_task=None,
    alter_type=None,
):
    """
    :param task_name: task名
    :param timedelta: 子任务颗粒度
    :param config_text: 详细配置
    :param task_rank: 任务排序,从1开始
    :param start_time: 开始时间
    :param end_time: 结束时间
    :param depend_task: 依赖的任务名
    :param alter_type: 更改
    :return:
    """
    task_name += subfix
    if depend_task:
        depend_task = ",".join(
            [task + subfix for task in depend_task.split(",")]
        )
    if timedelta.split(" ")[1] not in [
        "min",
        "H",
        "d",
        "w-mon",
        "MS",
        "QS",
        "YS",
    ]:
        raise PermissionError
    if alter_type not in ["must", "add_indicators", None]:
        raise PermissionError
    select_sql = (
        "select coalesce(max(task_id), 0) + 1 task_id from task_config"
    )
    result = mysql_con.get(select_sql)
    print(result)
    task_id = result[0]["task_id"]
    sql = "insert into task_config(task_id, task_name, timedelta, config_text, depend_task, task_rank, alter_type, start_time, end_time) values(%s, %s, %s, %s, %s, %s, %s, %s, %s)"
    mysql_con.execute(
        [
            {
                "sql": sql,
                "value": [
                    task_id,
                    task_name,
                    timedelta,
                    json.dumps(config_text),
                    depend_task,
                    task_rank,
                    alter_type,
                    start_time,
                    end_time,
                ],
            }
        ]
    )


def truncate_table():
    sql = "truncate table task_config"
    mysql_con.get(sql)
    sql_2 = "truncate table task_log"
    mysql_con.get(sql_2)


if __name__ == "__main__":
    start_time = "2019-01-01"
    end_time = None
    tasks = config["gen_task_name"]
    first = False
    subfix = "_" + str(datetime.datetime.now().replace(microsecond=0))
    no_excute = False

    opts, args = getopt.getopt(sys.argv[1:], "s:e:t:fn", ["kt"])
    for opt, val in opts:
        if opt == "-s":
            start_time = val
        elif opt == "-e":
            end_time = val
        elif opt == "-t":
            tasks = val.split(",")
        elif opt == "-f":
            first = True
        elif opt == "-n":
            no_excute = True

    if no_excute:  # 什么都不做
        exit()
    if first:
        truncate_table()
    fisrt_end_time = (
        datetime.datetime.strptime(start_time, "%Y-%m-%d")
        + relativedelta(hours=1)
    ).strftime("%Y-%m-%d %H:%M:%S")
    if "task_first" in tasks:
        main(
            "task_first",
            "1 d",
            task_first,
            1,
            start_time=start_time,
            end_time=fisrt_end_time,
            alter_type="must",
        )
    if "task_base" in tasks:
        main(
            "task_base",
            "15 min",
            task_base,
            2,
            start_time=start_time,
            end_time=end_time,
        )
    if "task_low" in tasks:
        main(
            "task_low",
            "15 min",
            task_low,
            2,
            start_time=start_time,
            end_time=end_time,
        )
    if "task_industry_agg" in tasks:
        main(
            "task_industry_agg",
            "15 min",
            task_industry_agg,
            3,
            start_time=start_time,
            end_time=end_time,
            depend_task="task_base",
        )
    if "task_baseline" in tasks:
        main(
            "task_baseline",
            "15 min",
            task_baseline,
            2,
            start_time=start_time,
            end_time=end_time,
        )
    if "task_org_day" in tasks:
        main(
            "task_org_day",
            "1 d",
            task_org_day,
            4,
            start_time=start_time,
            end_time=end_time,
            depend_task="task_industry_agg",
        )
    if "task_org_type_day" in tasks:
        main(
            "task_org_type_day",
            "1 d",
            task_org_type_day,
            4,
            start_time=start_time,
            end_time=end_time,
            depend_task="task_base",
        )
    if "task_day_sichuan" in tasks:
        main(
            "task_day_sichuan",
            "1 d",
            task_day_sichuan,
            3,
            start_time=start_time,
            end_time=end_time,
            depend_task="task_base",
        )
    if "task_year" in tasks:
        if end_time is None:
            tmp_end_time = datetime.datetime.now().replace(
                hour=0, minute=0, second=0, microsecond=0
            )
        else:
            tmp_end_time = datetime.datetime.strptime(end_time, "%Y-%m-%d")
        tmp_start_time = datetime.datetime.strptime(start_time, "%Y-%m-%d")
        if tmp_end_time.year == tmp_start_time.year:
            main(
                "task_year",
                f"{max((tmp_end_time.replace(day=1)-relativedelta(days=1) - tmp_start_time).days, 1)} d",
                task_year,
                4,
                start_time=start_time,
                end_time=str(
                    tmp_end_time.replace(day=1) - relativedelta(days=2)
                ),
            )
        else:
            main(
                "task_year",
                "1 YS",
                task_year,
                4,
                start_time=start_time,
                end_time=str(tmp_end_time.replace(year=tmp_end_time.year - 1)),
            )
            main(
                "task_year_now",
                f"{max(tmp_end_time.timetuple().tm_yday - 2, 1)} d",
                task_year,
                4,
                start_time=str(
                    tmp_start_time.replace(
                        year=tmp_end_time.year, month=1, day=1
                    )
                ),
                end_time=str(tmp_end_time - relativedelta(days=2)),
            )
    if "task_modify_alias" in tasks:
        main(
            "task_modify_alias",
            "1 d",
            task_modify_alias,
            4,
            start_time=start_time,
            end_time=end_time,
        )
    if "task_cons_day" in tasks:
        main(
            "task_cons_day",
            "1 d",
            task_cons_day,
            2,
            start_time=start_time,
            end_time=end_time,
        )
    if "task_cons_day_stat" in tasks:
        main(
            "task_cons_day_stat",
            "1 d",
            task_cons_day_stat,
            2,
            start_time=start_time,
            end_time=end_time,
        )
    if "task_cons_sort" in tasks:
        main(
            "task_cons_sort",
            "1 d",
            task_cons_sort,
            2,
            start_time=str(
                datetime.datetime.now().replace(microsecond=0)
                - relativedelta(hours=1)
            ),
            end_time=str(datetime.datetime.now().replace(microsecond=0)),
        )
    if "task_orderly" in tasks:
        main(
            "task_orderly",
            "15 min",
            task_orderly,
            2,
            start_time=start_time,
            end_time=end_time,
        )
    if "task_demand" in tasks:
        main(
            "task_demand",
            "15 min",
            task_demand,
            2,
            start_time=start_time,
            end_time=end_time,
        )
